package com.chenjj.bigdata.flink.window.function

import org.apache.flink.api.common.functions.AggregateFunction

/**
  * 定义求取平均值的Aggregation
  */
class MyAverageAggregation extends AggregateFunction[(String,Long),(Long,Long),Double]{

  /**
    * 定义createAccumulator为两个参数的元组，默认值为(0L,0L) ,第一个Long为累计值，第二个Long为累计数，平均值=第一个Long/第二个Long
    * @return
    */
  override def createAccumulator(): (Long, Long) = {
    (0L,0L)
  }

  /**
    *  定义输入数据累加到accumulator的逻辑
    * @param value  输入值
    * @param accumulator 累加器
    * @return
    */
  override def add(value: (String, Long), accumulator: (Long, Long)): (Long, Long) = {
    (accumulator._1 + value._2 , accumulator._2 +1)
  }

  /**
    * 获取平均值：平均值=第一个Long/第二个Long
    * @param accumulator
    * @return
    */
  override def getResult(accumulator: (Long, Long)): Double = {
    accumulator._1 / accumulator._2
  }

  /**
    * 定义累加器的合并逻辑
    * @param a
    * @param b
    * @return
    */
  override def merge(a: (Long, Long), b: (Long, Long)): (Long, Long) = {
    (a._1 + a._1,b._2+b._2)
  }
}
